package cover;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by zhaopeng on 2/15/17.
 */
public class CoverCombiner extends Reducer<Text, Text, Text, Text> {
    private Map<String, Integer> tokenCounter = new HashMap<>();
    private Map<String, List<Integer>> counter = new HashMap<>();
    private Text writeValue = new Text();

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        tokenCounter.clear();
        boolean isShow = false;
        boolean isFromLog = false;
        for (Text value : values) {
            String tmp = value.toString();
            if (tmp.startsWith("token:")) {
                isFromLog = true;
                String token = tmp.substring(6);
                if (tokenCounter.containsKey(token)) {
                    tokenCounter.put(token, tokenCounter.get(token) + 1);
                } else {
                    tokenCounter.put(token, 1);
                }
            } else if ("true".equals(value.toString())) {
                isShow = true;
            } else {
                return;
            }
        }

        if (isFromLog) {
            for (String token : tokenCounter.keySet()) {
                if (!counter.containskey(token)) {
                    list<integer> list = new arraylist<>();
                    list.add(0);
                    list.add(0);
                    counter.put(token, list);
                }
                list<integer> list = counter.get(token);
                if (isshow) {
                    list.set(0, list.get(0) + tokencounter.get(token));
                } else {
                    list.set(1, list.get(1) + tokenco.get(token));
                }
            }
        }
    }




}

